package com.bop.appender;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import java.io.IOException;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.List;

/**
 * @Author: LiFuqiang
 * @Date: 2022/1/21 17:03
 * @Version 1.0
 * @description 需要有空闲文件连接检测机制，检测当前文件连接是否长时间没有写入数据。
 */
public class HdfsWriter {
    private String dir;
    private String fileName;
    private int rollInterval; //  以 ms 为单位
    private long lastWriteTime;//  以 ms 为单位
    private long lastFileTime;//  以 ms 为单位
    private FSDataOutputStream currentOutStream;
    private String currentFile;
    private long maxIdleTime; //  以 ms 为单位
    private boolean isClosed = false;

    public HdfsWriter() {
        this.dir = "webhdfs://10.203.10.252:50070/tmp/test";
        this.fileName = "sample.log";
        this.lastWriteTime = System.currentTimeMillis();
        this.rollInterval = 60000;
        this.maxIdleTime = 5000;
        createNewFile();
    }

    public HdfsWriter(String dir, String fileName, int rollInterval, long maxIdleTime) {
        this.dir = dir;
        this.fileName = fileName;
        this.rollInterval = rollInterval;
        this.maxIdleTime = maxIdleTime;
        this.lastWriteTime = System.currentTimeMillis();
        createNewFile();
    }

    public synchronized void write(List<byte[]> events) {

        if (events.size() == 0) {
            if (isMaxIdle()) {
                // 如果达到空闲时间，则关闭文件
                close();
            }
            return;
        }

        if (shouldRoll()) {
            // 写入新文件
            createNewFile();
        }
        if (isClosed) {
            reopen(new Configuration());
        }
        try {
            events.forEach(this::writeData);
            this.currentOutStream.flush();
        } catch (IOException e) {
            e.printStackTrace();
        }
        this.lastWriteTime = System.currentTimeMillis();
    }

    public void reopen(Configuration conf) {
        try {
            Path dstPath = new Path(currentFile);
            FileSystem hdfs = getDfs(conf, dstPath);
            if (conf.getBoolean("dfs.append.support", true) && hdfs.isFile(dstPath)) {
                this.currentOutStream = hdfs.append(dstPath);
                this.isClosed = false;
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public void createNewFile() {
        try {
            if (isClosed == false && this.currentOutStream != null) {
                this.currentOutStream.close();
            }
            Configuration conf = new Configuration();
            this.currentFile = generateFileName();
            Path dstPath = new Path(currentFile);
            FileSystem hdfs = getDfs(conf, dstPath);
            this.currentOutStream = hdfs.create(dstPath);
            this.lastFileTime = System.currentTimeMillis();
            this.isClosed = false;
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public String generateFileName() {
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyyMMddHHmmss");
        String date = LocalDateTime.now().format(formatter);
        return this.dir + Path.SEPARATOR + fileName + "." + date;
    }

    protected FileSystem getDfs(Configuration conf, Path dstPath) throws IOException {
        return dstPath.getFileSystem(conf);
    }

    public void writeData(byte[] data) {
        try {
            this.currentOutStream.write(data);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public boolean shouldRoll() {
        long currentTime = System.currentTimeMillis();
        long interval = currentTime - this.lastFileTime;
        return interval >= this.rollInterval;
    }

    public boolean isMaxIdle() {
        long currentTime = System.currentTimeMillis();
        long interval = currentTime - this.lastWriteTime;
        return interval >= this.maxIdleTime;
    }

    public void close() {
        if (isClosed) return;
        try {
            this.currentOutStream.close();
            this.currentOutStream.hflush();
        } catch (IOException e) {
            e.printStackTrace();
        }
        this.isClosed = true;
    }

    public String getDir() {
        return dir;
    }

    public void setDir(String dir) {
        this.dir = dir;
    }

    public String getFileName() {
        return fileName;
    }

    public void setFileName(String fileName) {
        this.fileName = fileName;
    }

    public int getRollInterval() {
        return rollInterval;
    }

    public void setRollInterval(int rollInterval) {
        this.rollInterval = rollInterval;
    }

    public long getLastWriteTime() {
        return lastWriteTime;
    }

    public void setLastWriteTime(long lastWriteTime) {
        this.lastWriteTime = lastWriteTime;
    }

    public long getMaxIdleTime() {
        return maxIdleTime;
    }

    public void setMaxIdleTime(long maxIdleTime) {
        this.maxIdleTime = maxIdleTime;
    }
}
